package org.shj.spark.streaming;

import java.util.ArrayList;
import java.util.List;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.Optional;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.api.java.function.PairFunction;
import org.apache.spark.streaming.Durations;
import org.apache.spark.streaming.api.java.JavaPairDStream;
import org.apache.spark.streaming.api.java.JavaReceiverInputDStream;
import org.apache.spark.streaming.api.java.JavaStreamingContext;

import scala.Tuple2;

public class TransferOperation {

	public static void main(String[] args) throws Exception {
		SparkConf conf = new SparkConf().setAppName("TransferOperation").setMaster("local[2]");
		JavaStreamingContext jsc = new JavaStreamingContext(conf, Durations.seconds(5));
		
		//模拟一个黑名单，true代表是黑名单里的人
		List<Tuple2<String, Boolean>> blackList = new ArrayList<Tuple2<String, Boolean>>();
		blackList.add(new Tuple2<String, Boolean>("xuruyau", false));
		blackList.add(new Tuple2<String, Boolean>("xiaoriben", true));
		
		final JavaPairRDD<String, Boolean> blackRDD = jsc.sparkContext().parallelizePairs(blackList);
		JavaReceiverInputDStream<String> messageDStream = jsc.socketTextStream("ubuntu2", 8888);
		
		JavaPairDStream<String, String> pairDStream = messageDStream.mapToPair(new PairFunction<String, String, String>() {
			private static final long serialVersionUID = -1171783504990328891L;

			//line 格式为： 发送者 + " " + 接收者 + 消息
			public Tuple2<String, String> call(String line) throws Exception {
				return new Tuple2<String, String>(line.split(" ")[0], line);
			}
		});
		
		pairDStream.transform(new Function<JavaPairRDD<String,String>, JavaRDD<String>>() {
			private static final long serialVersionUID = -8467701807293550740L;

			public JavaRDD<String> call(JavaPairRDD<String, String> msgRDD) throws Exception {
				
				//返回的<String, Tuple2<String, Optional<Boolean>>>中，第一个String是两个RDD中匹配的key
				//Tuple2<String, Optional<Boolean>>中，
				//       String 表示msgRDD中key对应的value，
				//       Optional<Boolean> 表示blackRDD 中key对应的value     
				JavaPairRDD<String, Tuple2<String, Optional<Boolean>>> joinedRDD = 
						msgRDD.leftOuterJoin(blackRDD);
				JavaPairRDD<String, Tuple2<String, Optional<Boolean>>> filterRDD = 
						joinedRDD.filter(new Function<Tuple2<String,Tuple2<String,Optional<Boolean>>>, Boolean>() {
							private static final long serialVersionUID = -8179260544664094656L;
		
							@Override
							public Boolean call(Tuple2<String, Tuple2<String, Optional<Boolean>>> tuple) throws Exception {
								//isPresent() 为True表示 Optional是Some，表示匹配到了
								if(tuple._2._2.isPresent() && tuple._2._2.get()){
									return false;
								}
								return true;
							}
				});
				
				JavaRDD<String> validRDD = filterRDD.map(new Function<Tuple2<String,Tuple2<String,Optional<Boolean>>>, String>() {
					private static final long serialVersionUID = -1105284170900909580L;

					public String call(Tuple2<String, Tuple2<String, Optional<Boolean>>> v1) throws Exception {
						return v1._2._1;
					}
				});
				
				return validRDD;
			}
		});
		
		jsc.start();
		jsc.awaitTermination();
		jsc.close();

	}

}
